package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.bean.VisitorStats;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class ProductStatsApp {
    public static void main(String[] args)  throws Exception{
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);


        //TODO 1 从kafka中获取数据流
        String groupId = "product_stats_app";
        String pageViewSourceTopic="dwd_page_log";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSourceTopic = "dwm_payment_wide";
        String cartInfoSourceTopic = "dwd_cart_info";
        String favorInfoSourceTopic = "dwd_favor_info";
        String refundInfoSourceTopic = "dwd_order_refund_info";
        String commentInfoSourceTopic = "dwd_comment_info";

        FlinkKafkaConsumer<String> pageViewSource = MyKafkaUtil.getKafkaSource(pageViewSourceTopic,groupId);
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideSourceTopic,groupId);
        FlinkKafkaConsumer<String> paymentWideSource = MyKafkaUtil.getKafkaSource(paymentWideSourceTopic,groupId);
        FlinkKafkaConsumer<String> favorInfoSource = MyKafkaUtil.getKafkaSource(favorInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> cartInfoSource = MyKafkaUtil.getKafkaSource(cartInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> refundInfoSource = MyKafkaUtil.getKafkaSource(refundInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> commentInfoSource = MyKafkaUtil.getKafkaSource(commentInfoSourceTopic,groupId);

        DataStreamSource<String> pageViewDS = env.addSource(pageViewSource);
        DataStreamSource<String> orderWideDS = env.addSource(orderWideSource);
        DataStreamSource<String> paymentWideDS = env.addSource(paymentWideSource);
        DataStreamSource<String> favorInfoDS = env.addSource(favorInfoSource);
        DataStreamSource<String> cartInfoDS = env.addSource(cartInfoSource);
        DataStreamSource<String> refundInfoDS = env.addSource(refundInfoSource);
        DataStreamSource<String> commentInfoDS = env.addSource(commentInfoSource);


        //TODO 2  将获取的数据流进行结构的转换
        // 2.1 转换曝光及页面流数据
        SingleOutputStreamOperator<ProductStats> pageViewStatsDS = pageViewDS.process(new ProcessFunction<String, ProductStats>() {

            @Override
            public void processElement(String value, Context ctx, Collector<ProductStats> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject pageJsonObj = jsonObject.getJSONObject("page");
                String pageId = pageJsonObj.getString("page_id");
                if (pageId == null) {
                    System.out.println(jsonObject);
                }

                Long ts = jsonObject.getLong("ts");
                assert pageId != null;
                if (pageId.equals("good_detail")) {
                    Long skuId = pageJsonObj.getLong("item");
                    ProductStats productStats = ProductStats.builder().sku_id(skuId)
                            .click_ct(1L).ts(ts).build();
                    out.collect(productStats);
                }

                //曝光
                JSONArray displays = jsonObject.getJSONArray("displays");
                if (displays != null && displays.size() > 0) {
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject display = displays.getJSONObject(i);
                        if (display.getString("item_type").equals("sku_id")) {
                            Long skuId = display.getLong("item");
                            ProductStats productStats = ProductStats.builder().sku_id(skuId)
                                    .display_ct(1L).ts(ts).build();
                            out.collect(productStats);
                        }
                    }
                }

            }
        });

        //2.2 转换下单流数据
        SingleOutputStreamOperator<ProductStats> orderWideStatsDS = orderWideDS.map(
                json -> {
                    OrderWide orderWide = JSON.parseObject(json, OrderWide.class);
                    System.out.println("orderWide: ===== " + orderWide);
                    String create_time = orderWide.getCreate_time();
                    Long ts = DateTimeUtil.toTs(create_time);
                    return ProductStats.builder().sku_id(orderWide.getSku_id())
                            .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                            .order_sku_num(orderWide.getSku_num())
                            .order_amount(orderWide.getSplit_total_amount())
                            .ts(ts)
                            .build();
                }
        );
        //2.3 收藏转换数据
        SingleOutputStreamOperator<ProductStats> favorInfoStatsDS = favorInfoDS.map(
                json -> {
                    JSONObject favorInfo = JSON.parseObject(json);
                    Long ts = DateTimeUtil.toTs(favorInfo.getString("create_time"));
                    return ProductStats.builder().sku_id(favorInfo.getLong("sku_id"))
                            .favor_ct(1L).ts(ts).build();
                }
        );

        //2.4 购物车数据
        SingleOutputStreamOperator<ProductStats> cartInfoStatsDS = cartInfoDS.map(json -> {
            JSONObject cartInfo = JSON.parseObject(json);
            Long ts = DateTimeUtil.toTs(cartInfo.getString("create_time"));
            return ProductStats.builder().sku_id(cartInfo.getLong("sku_id"))
                    .cart_ct(1L).ts(ts).build();
        });

        //2.5 退款
        SingleOutputStreamOperator<ProductStats> refundInfoStatsDS = refundInfoDS.map(json -> {
            JSONObject refundInfo = JSON.parseObject(json);
            Long ts = DateTimeUtil.toTs(refundInfo.getString("create_time"));
            return ProductStats.builder()
                    .sku_id(refundInfo.getLong("sku_id"))
                    .refund_amount(refundInfo.getBigDecimal("refund_amount"))
                    .refundOrderIdSet(
                            new HashSet(Collections.singleton(refundInfo.getLong("order_id")))
                    )
                    .ts(ts)
                    .build();
        });

        //2.6 支付
        SingleOutputStreamOperator<ProductStats> paymentStatsDS = paymentWideDS.map(
                json -> {
                    PaymentWide paymentWide = JSON.parseObject(json, PaymentWide.class);
                    Long ts = DateTimeUtil.toTs(paymentWide.getPayment_create_time());
                    return ProductStats.builder().sku_id(paymentWide.getSku_id())
                            .payment_amount(paymentWide.getTotal_amount())
                            .paidOrderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                            .ts(ts)
                            .build();
                }
        );



        // 2.7 评价
        SingleOutputStreamOperator<ProductStats> commentInfoStatsDS = commentInfoDS.map(json -> {
            JSONObject commentInfo = JSON.parseObject(json);
            Long ts = DateTimeUtil.toTs(commentInfo.getString("create_time"));
            Long goodCt = GmallConstant.APPRAISE_GOOD.equals(commentInfo.getString("appraise")) ? 1L : 0L;
            return ProductStats.builder()
                    .sku_id(commentInfo.getLong("sku_id"))
                    .comment_ct(1L)
                    .good_comment_ct(goodCt)
                    .ts(ts)
                    .build();
        });


        //TODO 3 将统一的数据结构流合并为一个流
        DataStream<ProductStats> productStatsDetailDS = pageViewStatsDS.union(
                orderWideStatsDS,
                cartInfoStatsDS,
                paymentStatsDS,
                refundInfoStatsDS,
                favorInfoStatsDS,
                commentInfoStatsDS
        );

        //TODO 4 .设定事件时间与水位线
        SingleOutputStreamOperator<ProductStats> productStatsWithTsDS= productStatsDetailDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ProductStats>forMonotonousTimestamps()
                        .withTimestampAssigner((SerializableTimestampAssigner<ProductStats>) (productStats, l) -> productStats.getTs())
        );


        //TODO 5. 分组，开窗，聚合
        SingleOutputStreamOperator<ProductStats> productStatsDs= productStatsWithTsDS
                .keyBy((KeySelector<ProductStats, Long>) ProductStats::getSku_id)
        //5.2 开窗
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // reduce聚合
        .reduce(new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats p1, ProductStats p2) throws Exception {
                        p1.setDisplay_ct(p1.getDisplay_ct() + p2.getDisplay_ct());
                        p1.setClick_ct(p1.getClick_ct() + p2.getClick_ct());
                        p1.setCart_ct(p1.getCart_ct() + p2.getCart_ct());
                        p1.setFavor_ct(p1.getFavor_ct() + p2.getFavor_ct());
                        p1.setOrder_amount(p1.getOrder_amount().add(p2.getOrder_amount()));
                        p1.getOrderIdSet().addAll(p1.getOrderIdSet());
                        p1.setOrder_ct((long) p1.getOrderIdSet().size());
                        p1.setOrder_sku_num(p1.getOrder_sku_num() + p2.getOrder_sku_num());

                        p1.setPayment_amount(p1.getPayment_amount().add(p2.getPayment_amount()));
                        p1.getRefundOrderIdSet().addAll(p2.getRefundOrderIdSet());
                        p1.setRefund_order_ct((long) p1.getRefundOrderIdSet().size());
                        p1.setRefund_amount(p1.getRefund_amount().add(p2.getRefund_amount()));

                        p1.getPaidOrderIdSet().addAll(p2.getPaidOrderIdSet());
                        p1.setPaid_order_ct((long) p1.getPaidOrderIdSet().size());

                        p1.setComment_ct(p1.getComment_ct() + p2.getComment_ct());
                        p1.setGood_comment_ct(p1.getGood_comment_ct() + p2.getGood_comment_ct());
                        return p1;
                    }
                },
                new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void process(Long aLong, Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                        for(ProductStats productStats: elements){
                            productStats.setStt(simpleDateFormat.format(context.window().getStart()));
                            productStats.setEdt(simpleDateFormat.format(context.window().getEnd()));
                            //TS的设置
                            productStats.setTs(new Date().getTime());
                            out.collect(productStats);
                        }
                    }
                }

        );


        //TODO 6 补充商品维度信息
        // 6.1 补充SKU维度  getKey用于从流中获取主键
        SingleOutputStreamOperator<ProductStats> productStatsWithSkuDS = AsyncDataStream.unorderedWait(productStatsDs, new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {
            // productStats 是流对象 ； jsonObject 是从Hbase中获取的数据，只不过转换为了JSONObject对象
            @Override
            public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                productStats.setSku_name(jsonObject.getString("SKU_NAME"));
                productStats.setSku_price(jsonObject.getBigDecimal("PRICE"));
                productStats.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                productStats.setSpu_id(jsonObject.getLong("SPU_ID"));
                productStats.setTm_id(jsonObject.getLong("TM_ID"));
            }

            @Override
            public String getKey(ProductStats productStats) {
                return String.valueOf(productStats.getSku_id());
            }
        }, 60, TimeUnit.SECONDS);


        // 6.2 补充spu维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSPuDS = AsyncDataStream.unorderedWait(productStatsWithSkuDS,
                new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(ProductStats productStats) {
                        return String.valueOf(productStats.getSpu_id());
                    }
                }, 60, TimeUnit.SECONDS);


        // 6.3 补充品类维度
        SingleOutputStreamOperator<ProductStats> productStatsWithCategoryDS = AsyncDataStream.unorderedWait(
                productStatsWithSPuDS,
                new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setCategory3_name(jsonObject.getString("NAME"));
                    }

                    @Override
                    public String getKey(ProductStats productStats) {
                        return String.valueOf(productStats.getCategory3_id());
                    }
                }, 60, TimeUnit.SECONDS
        );
        //6.4 补充品牌维度
        SingleOutputStreamOperator<ProductStats> productStatsWithTmDs = AsyncDataStream.unorderedWait(productStatsWithCategoryDS,
                new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(ProductStats productStats, JSONObject jsonObject) throws Exception {
                        productStats.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(ProductStats productStats) {
                        return String.valueOf(productStats.getTm_id());
                    }
                }, 60, TimeUnit.SECONDS);



        //TODO 7 写入到ClickHouse
        productStatsWithTmDs.addSink(
                ClickHouseUtil.getJdbcSink(
                        "insert into product_stats_2021 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
                )
        );



        env.execute();

    }
}
